package com.shelpe.services.tenant.controller;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import com.shelpe.services.tenant.dto.MessageDto;
import com.shelpe.services.tenant.model.PurchaseOrder;
import com.shelpe.services.tenant.model.Users;
import com.shelpe.services.tenant.proxy.MessageServiceClient;
import com.shelpe.services.tenant.proxy.MessageServiceClient.CancelResponse;
import com.shelpe.services.tenant.proxy.MessageServiceClient.ConsumeMessageResponse;
import com.shelpe.services.tenant.proxy.MessageServiceClient.PermitResponse;
import com.shelpe.services.tenant.proxy.MessageServiceClient.StatusRes;
import com.shelpe.services.tenant.proxy.MessageServiceClient.TmcMessage;
import com.shelpe.services.tenant.service.OrderService;
import com.shelpe.services.tenant.service.UserService;

@RestController
@RequestMapping(value="/jobserver")
public class MessageController {

	private static final Logger logger = LoggerFactory.getLogger(MessageController.class);
	
	@Autowired
	private MessageServiceClient messageClient;
	
	@Autowired
	private UserService userService;
	
	@Autowired
	private OrderService orderService;
	
	@Value("${jobserver.appKeyUserId}")
	private String appKeyUserId;
	
	private final ExecutorService executor = Executors.newSingleThreadExecutor();
	
	@RequestMapping("/messagepermit")
	@ResponseBody
	public MessageDto permitMessage(@RequestParam("user") String userId, @RequestParam("topics") String topics){
		
		MessageDto dto = new MessageDto();
		
		String[] innerUserId = appKeyUserId.split(",");
		
		if(userId != null){
			
			innerUserId = new String[]{userId};
		}
		
		if(innerUserId != null && innerUserId.length > 0){
			
			for(int i = 0; i < innerUserId.length; i++){
				
				PermitResponse res = messageClient.permitMessage(Long.valueOf(innerUserId[i]), topics);
				
				if(res != null && res.getRes() != null){
					StatusRes status = res.getRes();
					dto.setStatus(status.getIsSuccess());
					logger.info(">>>permitMessage: userId {} Success", Long.valueOf(innerUserId[i]));
				}else{
					dto.setStatus(false);
					if(res != null && res.getError() != null){
						dto.setMessage(res.getError().getSubMessage());
					}
					logger.info(">>>permitMessage: userId {} Failed, error {}", Long.valueOf(innerUserId[i]), res.getError().getSubMessage());
				}
			}
		}
		
		return dto;	
	}
	@RequestMapping("/messagecancel")
	@ResponseBody
	public MessageDto cancelMessage(@RequestParam("user") Long userId){
		
		MessageDto dto = new MessageDto();
		Users user = userService.getUserByUserId(userId);
		CancelResponse res = messageClient.cancelMessage(user.getNick());
		
		if(res != null && res.getRes() != null){
			StatusRes status = res.getRes();
			dto.setStatus(status.getIsSuccess());
			logger.info(">>>cancelMessage: userId {} Success", userId);
		}else{
			dto.setStatus(false);
			if(res != null && res.getError() != null){
				dto.setMessage(res.getError().getSubMessage());
			}
			logger.info(">>>cancelMessage: userId {} Failed, error {}", userId, res.getError().getSubMessage());
		}
		
		return dto;
	}
	
	@RequestMapping("/messageconsume")
	@ResponseBody
	public MessageDto consumeMessage(){
		
		MessageDto dto = new MessageDto();
		
		CompletionService<MessageDto> comletionService = new ExecutorCompletionService<MessageDto>(executor); 
		
		comletionService.submit(new Callable<MessageDto>(){

			@Override
			public MessageDto call() throws Exception {
				// TODO Auto-generated method stub
				
				String[] innerUserId = appKeyUserId.split(",");
		
				if(innerUserId != null && innerUserId.length > 0){
					for(int i = 0; i < innerUserId.length; i++){
						
						ConsumeMessageResponse res = messageClient.consumeMessage(Long.valueOf(innerUserId[i]));
						if(res != null && res.getRes() != null && res.getRes().getMessages() != null){
							
							List<TmcMessage> tmcList = res.getRes().getMessages().getTmcList();
						
							if(tmcList != null && tmcList.size() > 0){
								
								List<Long> ids = new ArrayList<Long>();
								for(TmcMessage tmc : tmcList){
									if(processOrderContent(tmc)){
										ids.add(tmc.getId());
									}
								}
								
								 String messageIds = transIdsToStr(ids);
								 if(messageIds != null && messageIds.length() > 0){
									 messageClient.confirmMessage(Long.valueOf(innerUserId[i]), messageIds);
								 }
							}
							
						}else{
							if(res != null && res.getError() != null){
								logger.warn(">>>consumeMessage: error {}", res.getError().getSubMessage());
							}
						}
					}
				}
				
				return null;
			}
		});
		
		dto.setMessage("success");
		
		return dto;	
	}
	
	private boolean processOrderContent(TmcMessage tmc){
		
		boolean result = true;
		if(tmc.getTopic() != null && tmc.getTopic().equals("taobao_fuwu_OrderPaid")){
			
			PurchaseOrder order = orderService.parseOrderContent(tmc.getContent());
			int count = orderService.countById(order.getOrderId());
			if(count == 0){
				
				int addNum = orderService.insertOrder(order);
				result = (addNum == 1);
			}
		}
		
		return result;
	}
	
	private String transIdsToStr(List<Long> ids){
		
		StringBuilder sb = new StringBuilder();
		
		if(ids != null && ids.size() > 0){
			
			for(int i = 0; i < ids.size() - 1; i++){
				sb.append(ids.get(i));
				sb.append(",");
			}
			sb.append(ids.get(ids.size() - 1));
		}
		
		return sb.toString();
	}
	
	
}
